package com.mjk.common.tools.lock.distrib.impl;


import com.mjk.common.base.error.BizException;
import com.mjk.common.core.dcc.DccClient;
import com.mjk.common.tools.lock.distrib.DistributedLock;
import com.mjk.common.tools.logger.BizLogger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;


/**
 * 基于ZK的分布式锁
 * @author mjk
 * 对于大并发情况，有待考究，内存问题已解决
 */
public class ZookepperDistributedLock implements DistributedLock {
	private  String root = "/distrib";
	private String lockName;
	private ZooKeeper zk;
	private static final int SESSION_TIMEOUT = 10000;
	private String myZnode;
	private String waitNode;		
	public ZookepperDistributedLock(DccClient client, String lockName){
		try{
			this.lockName = lockName;
			zk = client.getZk();
			//初始化根节点
			Stat stat = zk.exists(root, false);
			if (stat == null) {
				this.zk.create(root, new byte[0],
						ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}			
		}catch(Exception e){
			if(null!=this.zk){
				try {
					zk.close();
				} catch (InterruptedException e1) {					
					Thread.currentThread().interrupt();
				}
			}
			BizLogger.error(e);
		}
	}
		
	
	public boolean lock(String key) {		
		if (tryLock()) {			
			return true;
		}
		return waitForLock(this.waitNode);
	}
	
	private boolean waitForLock(String lower){
		try{
			Stat stat = this.zk.exists(this.root + "/" + lower, true);
			long statTime = System.currentTimeMillis();
			while(null!=stat){				
				BizLogger.debug(" waiting for " + this.root + "/" + lower);
				stat = this.zk.exists(this.root + "/" + lower, true);
				//超时时间设置
				if(System.currentTimeMillis() - statTime > ZookepperDistributedLock.SESSION_TIMEOUT && null!=stat){
						this.zk.delete(this.root + "/" + lower, -1);
				}
			}
			return true;
		}catch(Exception e){
			BizLogger.error(e);
			Thread.currentThread().interrupt();
		}
		return false;
	}
	/**
	 * 
	 * @return
	 */
	public boolean tryLock() {
		try {
			String splitStr = "_lock_";
			if (this.lockName.contains(splitStr)) {
				throw new BizException("lockName can not contains " + splitStr);
			}
			this.myZnode = this.zk.create(this.root + "/" + this.lockName
					+ splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
					CreateMode.EPHEMERAL_SEQUENTIAL);
			
			BizLogger.debug(myZnode + " is created ");			
			List<String> subNodes = this.zk.getChildren(this.root, false);
			
			List<String> lockObjNodes = new ArrayList<>();
			for (String node : subNodes) {
				String tmpNode = node.split(splitStr)[0];
				if (tmpNode.equals(this.lockName)) {
					lockObjNodes.add(node);
				}
			}
			Collections.sort(lockObjNodes);
			if (myZnode.equals(this.root + "/"+  lockObjNodes.get(0))) {
				BizLogger.info(myZnode + "=="+ lockObjNodes.get(0));
				return true;
			}
			String subMyZnode = myZnode.substring(myZnode.lastIndexOf("/") + 1);
			this.waitNode = lockObjNodes.get(Collections.binarySearch(lockObjNodes, subMyZnode) - 1);
		} catch (Exception e) {
			BizLogger.error(e);
			Thread.currentThread().interrupt();
		}
		return false;
	}
	
	public void releaseLock(String key) {
		try {
			BizLogger.debug("unlock " + this.myZnode);
			this.zk.delete(this.myZnode, -1);
        } catch (Exception e) {
			BizLogger.error(e);
			Thread.currentThread().interrupt();
		}
    }

}
